package org.example.client;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.example.ServerConnection;
import org.example.commom.ZKConnectionHelp;
import org.example.protocol.RpcRequest;
import org.example.protocol.RpcResponse;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * rpc的调用者
 */
public class RpcInvoker {
    // 1、创建一个线程池对象，处理自定义事件
    private final ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    // 定义一个Map，保存<ip:port,连接>的键值对
    private Map<String, ServerConnection> serverMap;

    public RpcInvoker() throws InterruptedException {
        initServerConnection();
    }

    // 3、初始化客户端(创建连接池，创建bootstrap，设置bootstrap，连接服务器)
    public void initServerConnection() throws InterruptedException {
        serverMap = new ConcurrentHashMap<>();
        // 首先查询zk的/rpc/server下有多少个server，为每一个server创建连接
        ZkClient zk = new ZkClient(ZKConnectionHelp.zkUrl);
        if (zk.exists(ZKConnectionHelp.serverPath)) {
            List<String> children = zk.getChildren(ZKConnectionHelp.serverPath);

            // 遍历children列表，进行ServerConnection的创建
            for (String child : children) {
                System.out.println("new server: " + child + " found!");
                ServerConnection serverConnection = new ServerConnection(child);
                serverMap.put(child, serverConnection);
            }

            // 注册server节点的子节点监听
            zk.subscribeChildChanges(ZKConnectionHelp.serverPath, new IZkChildListener() {
                @Override
                public void handleChildChange(String parentPath, List<String> currentChildren) throws Exception {
                    Set<String> oldChildren = serverMap.keySet();
                    for (String child : currentChildren) {
                        if (!oldChildren.contains(child)) {
                            System.out.println("new server: " + child + " found!");
                            ServerConnection serverConnection = new ServerConnection(child);
                            serverMap.put(child, serverConnection);
                        }
                    }

                    for (String oldChild : oldChildren) {
                        if (!currentChildren.contains(oldChild)) {
                            System.out.println("lost server: " + oldChild + "!");
                            serverMap.remove(oldChild);
                        }
                    }

                }
            });
        }
    }

    // 4、编写一个方法，使用JDK的动态代理创建对象
    public Object createProxy(final Class<?> serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{serviceClass}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 2) rpcResponseHandler设置请求报文参数
                RpcRequest rpcRequest = new RpcRequest();
                rpcRequest.setRequestId("1");
                rpcRequest.setClassName(serviceClass.getName());
                rpcRequest.setMethodName(method.getName());
                rpcRequest.setParameterTypes(method.getParameterTypes());
                rpcRequest.setParameters(args);

                for (ServerConnection serverConnection : serverMap.values()) {
                    serverConnection.getRpcResponseHandler().setRpcRequest(rpcRequest);
                    // 3) 使用线程池来处理call()操作，并返回结果
                    try {
                        RpcResponse result = (RpcResponse) executorService.submit(serverConnection.getRpcResponseHandler()).get(500, TimeUnit.MILLISECONDS);
                        System.out.println("server: " + serverConnection.getIndex() + " get result: " + result.getResult());
                    } catch (Exception e) {
                        e.printStackTrace();
                        System.out.println("server: " + serverConnection.getIndex() + " timeout");
                    } finally {
                        serverConnection.getRpcResponseHandler().setRpcResponse(null);
                    }
                }

                // 4）返回结果
                return null;
            }
        });
    }

}
